package edu.zju.gis.dbfg.model.statistic;

import edu.zju.gis.dbfg.common.exception.ModelFailedException;
import edu.zju.gis.dbfg.model.PyramidConfig;
import edu.zju.gis.dbfg.model.datasource.DataReaderRDD;
import edu.zju.gis.dbfg.model.util.*;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.geotools.referencing.CRS;
import org.locationtech.jts.geom.Geometry;
import org.opengis.referencing.crs.CoordinateReferenceSystem;
import org.opengis.referencing.operation.MathTransform;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static edu.zju.gis.dbfg.model.base.Constant.DEFAULT_CRS;
import static edu.zju.gis.dbfg.model.base.Constant.DEFAULT_PYRAMID_CONFIG;
import static edu.zju.gis.dbfg.model.base.Constant.PYRAMID_CONFIG_4326;

public class LUContrastStatistic {
    private final static Logger logger = LoggerFactory.getLogger(LUContrastStatistic.class);
    public static void main (String[] args) throws Exception {
        logger.info("Setup Spark Context");
        SparkConf conf = new SparkConf();
        //conf.setMaster("local[1]");
        //conf.setAppName("lalala");
        JavaSparkContext jsc = new JavaSparkContext(conf);

        logger.info("Setup arguments");
        LUContrastStatisticArgs mArgs = LUContrastStatisticArgs.initArgs(args);
        if (mArgs == null) {
            String msg = "init args failed, exit";
            logger.error(msg);
            throw new ModelFailedException(LUContrastStatistic.class, "main()", msg, args);
        }

        CoordinateReferenceSystem crs1 = CRS.parseWKT(CrsPick2.pickByName(mArgs.getCrs1()));
        CoordinateReferenceSystem crs2 = CRS.parseWKT(CrsPick.pickByName(mArgs.getCrs2()));
        CoordinateReferenceSystem crs3 = CRS.parseWKT(CrsPick.pickByName(mArgs.getCrs3()));
        CoordinateReferenceSystem targetCRS = DEFAULT_CRS();
        //PyramidConfig targetPyramidConfig = DEFAULT_PYRAMID_CONFIG();
        PyramidConfig targetPyramidConfig = PYRAMID_CONFIG_4326();//gewang

        MathTransform transform1 = null;
        MathTransform transform2 = null;
        MathTransform transform3 = null;
        if (!crs1.equals(targetCRS)) {
            transform1 = CRS.findMathTransform(crs1, targetCRS);
        }
        if (!crs2.equals(targetCRS)) {
            transform2 = CRS.findMathTransform(crs2, targetCRS);
        }
        if(!crs3.equals(targetCRS)){
            transform3 = CRS.findMathTransform(crs3, targetCRS);
        }

        String input1 = mArgs.getInput1();
        String type1 = mArgs.getType1();
        Map<String, Object> attrs1 = new HashMap<>();
        JavaRDD<GeoItem> gi1 = DataReaderRDD.readToRDD(type1, jsc, input1, "1", attrs1, transform1);
        JavaPairRDD<String, GeoItem> gi1Pair = DataReaderRDD.mapToPair(gi1, targetPyramidConfig);

        String input2 = mArgs.getInput2();
        String type2 = mArgs.getType2();
        String dlbm2 = mArgs.getDlbmIndex2();
        String zldwdm2 = mArgs.getZldwdmIndex2();
        String area2 = mArgs.getAreaIndex2();
        String jarea2 = mArgs.getJareaIndex2();
        String kcdlbm2 = mArgs.getKcdlbmIndex2();

        Map<String, Object> attrs2 = new HashMap<>();
        attrs2.put("DLBM", dlbm2);
        attrs2.put("ZLDWDM", zldwdm2);
        attrs2.put("TBMJ", area2);
        attrs2.put("TBDLMJ", jarea2);
        attrs2.put("KCDLBM", kcdlbm2);
        JavaPairRDD<String, GeoItem> gi2Pair = null;

        if (type2.equals("ES")) {
            gi1Pair.cache();
            List<TileID> tileIDs = gi1Pair.map(t -> TileID.fromString(t._1)).distinct().collect();
            gi2Pair = DataReaderRDD.readToPairRDD(type2, jsc, input2, attrs2, tileIDs, targetPyramidConfig);
            //gi2Pair = DataReaderRDD.readToRDD2(type2, jsc, input2, attrs2, null, null, transform2);

        } else {
            JavaRDD<GeoItem> gi2 = DataReaderRDD.readToRDD(type2, jsc, input2, "1", attrs2, transform2);
            gi2Pair = DataReaderRDD.mapToPair(gi2, targetPyramidConfig);
        }

        String input3 = mArgs.getInput3();
        String type3 = mArgs.getType3();
        String dlbm3 = mArgs.getDlbmIndex3();
        String zldwdm3 = mArgs.getZldwdmIndex3();
        String area3 = mArgs.getAreaIndex3();
        Map<String, Object> attrs3 = new HashMap<>();
        attrs3.put("DLBM", dlbm3);
        attrs3.put("ZLDWDM", zldwdm3);
        attrs3.put("AREA",area3);
        JavaPairRDD<String, GeoItem> gi3Pair = null;

        if (type3.equals("ES")) {
        } else {
            JavaRDD<GeoItem> gi3 = DataReaderRDD.readToRDD(type3, jsc, input3, "1", attrs3, transform3);
            gi3Pair = DataReaderRDD.mapToPair(gi3, targetPyramidConfig);
        }

        JavaPairRDD<String, Tuple2<GeoItem, GeoItem>> joined2 = gi1Pair.join(gi2Pair);
        JavaPairRDD<String, GeoItem> tempRDD2 = joined2.mapToPair(new PairFunction<Tuple2<String, Tuple2<GeoItem, GeoItem>>, String, GeoItem>() {
            @Override
            public Tuple2<String, GeoItem> call(Tuple2<String, Tuple2<GeoItem, GeoItem>> input) throws Exception {
                GeoItem gi2 = input._2._2;
                GeoItem gi1 = input._2._1;
                Geometry g1 = input._2._1.getGeometry();
                Geometry g2 = input._2._2.getGeometry();
                if (!g1.intersects(g2)) {
                    return new Tuple2<>("EMPTY", null);
                }
                Geometry intersection = g1.intersection(g2);
                Double iarea = intersection.getArea();
                //Double oarea = g2.getArea();

                String dlbm =  gi2.getProp().get("DLBM").toString().replace("\"","").replace("\40","");
                String zldwdm = String.valueOf(gi2.getProp().get("ZLDWDM")).replace("\"","").replace("\40","").substring(0,9);
                Double tbdlmj=Double.valueOf(gi2.getProp().get("TBDLMJ").toString());
                String kcdlbm = String.valueOf(gi2.getProp().get("KCDLBM")).replace("\"","").replace("\40","");
                Double tbmj = Double.valueOf(gi2.getProp().get("TBMJ").toString());
                Double ytbmj = Double.valueOf(gi2.getProp().get("YTBMJ").toString());
                Double itbdlmj = iarea/ytbmj * tbdlmj;
                Double itbmj = iarea/ytbmj * tbmj;

                String id = zldwdm+"##"+dlbm;
                GeoItem result = new GeoItem();
                result.setFid(id);
                Map<String, Object> props = new HashMap<>();
                props.put("ZLDWDM", zldwdm);
                props.put("DLBM", dlbm);
                props.put("KCDLBM",kcdlbm);
                props.put("TBMJ",itbmj);
                props.put("AREA", itbdlmj);
                result.setProp(props);
                result.setGeometry(intersection);
                return new Tuple2<>(id, result);
            }
        }).filter(new Function<Tuple2<String, GeoItem>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, GeoItem> input) throws Exception {
                return !input._1.equals("EMPTY");
            }
        });

        tempRDD2.cache();
        //WriteWKT.writeToFile(tempRDD2, "G:\\GIT\\TestData\\contIntersect1.wkt");

        JavaPairRDD<String, GeoItem> kctempRDD = tempRDD2.mapToPair(new PairFunction<Tuple2<String, GeoItem>, String, GeoItem>() {
            @Override
            public Tuple2<String, GeoItem> call(Tuple2<String, GeoItem> input) throws Exception {
                String kcdlbm = String.valueOf(input._2.getProp().get("KCDLBM"));
                if(!kcdlbm.equals(null) && kcdlbm.length()>3){
                    String zldwdm = String.valueOf(input._2.getProp().get("ZLDWDM"));
                    Double kcmj = Double.valueOf(String.valueOf(input._2.getProp().get("TBMJ")))-Double.valueOf(String.valueOf(input._2.getProp().get("AREA").toString()));
                    String id = zldwdm + "##" + kcdlbm;
                    GeoItem result = new GeoItem();
                    result.setFid(id);
                    Map<String, Object> props = new HashMap<>();
                    props.put("DLBM", kcdlbm);
                    props.put("ZLDWDM", zldwdm);
                    props.put("AREA", kcmj);
                    props.put("KCDLBM","");
                    props.put("TBMJ","");

                    result.setProp(props);

                    return new Tuple2<>(id, result);
                }
                return new Tuple2<>("EMPTY", null);
            }
        }).filter(new Function<Tuple2<String, GeoItem>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, GeoItem> input) throws Exception {
                return !input._1.equals("EMPTY");
            }
        });
        kctempRDD.cache();

        JavaPairRDD<String, GeoItem> alltempRDD=tempRDD2.union(kctempRDD);


        JavaPairRDD<String, GeoItem> resultRDD2 = alltempRDD.reduceByKey(new Function2<GeoItem, GeoItem, GeoItem>() {
            @Override
            public GeoItem call(GeoItem gi1, GeoItem gi2) throws Exception {
                Double area = (Double) gi1.getProp().get("AREA") + (Double) gi2.getProp().get("AREA");
                GeoItem result = new GeoItem();
                result.setFid(gi1.getFid());
                Map<String, Object> props = new HashMap<>();
                props.put("AREA", area);
                result.setProp(props);
                return result;
            }
        });

        JavaPairRDD<String, Tuple2<GeoItem, GeoItem>> joined3 = gi1Pair.join(gi3Pair);
        JavaPairRDD<String, GeoItem> tempRDD3 = joined3.mapToPair(new PairFunction<Tuple2<String, Tuple2<GeoItem, GeoItem>>, String, GeoItem>() {
            @Override
            public Tuple2<String, GeoItem> call(Tuple2<String, Tuple2<GeoItem, GeoItem>> input) throws Exception {
                GeoItem gi2 = input._2._2;
                GeoItem gi1 = input._2._1;
                Geometry g1 = input._2._1.getGeometry();
                Geometry g2 = input._2._2.getGeometry();
                if (!g1.intersects(g2)) {
                    return new Tuple2<>("EMPTY", null);
                }
                Geometry intersection = g1.intersection(g2);
                Double area = intersection.getArea();
                String dlbm =  gi2.getProp().get("DLBM").toString().replace("\"","");
                String zldwdm = String.valueOf(gi2.getProp().get("ZLDWDM")).replace("\"","").substring(0,9);

                String id = zldwdm+"##"+dlbm;
                GeoItem result = new GeoItem();
                result.setFid(id);
                Map<String, Object> props = new HashMap<>();
                props.put("DLBM", dlbm);
                props.put("ZLDWDM", zldwdm);
                props.put("AREA", area);
                props.put("KCDLBM","0");
                props.put("TBMJ","0");
                result.setProp(props);
                result.setGeometry(intersection);
                return new Tuple2<>(id, result);
            }
        }).filter(new Function<Tuple2<String, GeoItem>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, GeoItem> input) throws Exception {
                return !input._1.equals("EMPTY");
            }
        });

        tempRDD3.cache();
        //WriteWKT.writeToFile(tempRDD3, "G:\\GIT\\TestData\\ContrastStat.wkt");

        JavaPairRDD<String, GeoItem> resultRDD3 = tempRDD3.reduceByKey(new Function2<GeoItem, GeoItem, GeoItem>() {
            @Override
            public GeoItem call(GeoItem gi1, GeoItem gi2) throws Exception {
                Double area = (Double) gi1.getProp().get("AREA") + (Double) gi2.getProp().get("AREA");
                GeoItem result = new GeoItem();
                result.setFid(gi1.getFid());
                Map<String, Object> props = new HashMap<>();
                props.put("AREA", area);
                result.setProp(props);
                return result;
            }
        });

        Map<String, GeoItem> result2 = resultRDD2.collectAsMap();
        Map<String, GeoItem> result3 = resultRDD3.collectAsMap();
        File file = new File(mArgs.getOutput());
        if (file.exists()) {
            file.delete();
        }
        FileWriter writer = new FileWriter(file);
        writer.write("Dataset1统计结果:\n");

        WriteToFile.writeToFile(writer,result2);

        writer.write("\nDataset2统计结果:\n");
        WriteToFile.writeToFile(writer,result3);
        writer.close();

    }
}
